package com.atguigu.edu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.app.BaseApp;
import com.atguigu.edu.bean.TradeProvinceOrderWindow;
import com.atguigu.edu.common.Constant;
import com.atguigu.edu.function.AsyncDimFunction;
import com.atguigu.edu.util.AtguiguUtil;
import com.atguigu.edu.util.DimUtil;
import com.atguigu.edu.util.FlinkSinkUtil;
import com.atguigu.edu.util.JdbcUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

public class Dws_11_Trade_Province extends BaseApp {
    public static void main(String[] args) {
        new Dws_11_Trade_Province().init(
                4011, 2,
                "Dws_11_Trade_Province",
                Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment ev, DataStreamSource<String> stream) {
        SingleOutputStreamOperator<TradeProvinceOrderWindow> streamWithoutDim = stream
                .map(new MapFunction<String, TradeProvinceOrderWindow>() {
                    @Override
                    public TradeProvinceOrderWindow map(String value) throws Exception {
                        JSONObject obj = JSON.parseObject(value);
                        return TradeProvinceOrderWindow.builder()
                                .provinceId(obj.getString("province_id"))
                                .orderDetailId(obj.getString("id"))
                                .orderAmount(obj.getBigDecimal("final_amount"))
                                .orderIdSet(new HashSet<>(Collections.singletonList(obj.getString("order_id"))))
                                .userIdSet(new HashSet<>(Collections.singletonList(obj.getString("user_id"))))
                                .ts(obj.getLong("ts") * 1000)
                                .build();
                    }

                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TradeProvinceOrderWindow>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                                .withIdleness(Duration.ofSeconds(6))
                )
                .keyBy(TradeProvinceOrderWindow::getOrderDetailId)
                .process(new KeyedProcessFunction<String, TradeProvinceOrderWindow, TradeProvinceOrderWindow>() {
                    private ValueState<TradeProvinceOrderWindow> beanState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        beanState = getRuntimeContext().getState(new ValueStateDescriptor<TradeProvinceOrderWindow>("bean", TradeProvinceOrderWindow.class));
                    }

                    @Override
                    public void processElement(TradeProvinceOrderWindow currentBean,
                                               Context ctx,
                                               Collector<TradeProvinceOrderWindow> out) throws Exception {
                        TradeProvinceOrderWindow lastBean = beanState.value();
                        if (lastBean == null) {
                            out.collect(currentBean);
                        } else {
                            lastBean.setOrderAmount(currentBean.getOrderAmount().subtract(lastBean.getOrderAmount()));
                            out.collect(lastBean);
                        }
                        //先把数据存入到状态中
                        beanState.update(currentBean);

                    }
                })
/*        1> TradeProvinceOrderWindow(orderDetailId=58787, orderIdSet=[53966], userIdSet=[47], stt=null, edt=null, provinceId=9,
 provinceName=, orderUserCount=null, orderCount=null, orderAmount=0.0, ts=1675230228000)*/
                .keyBy(TradeProvinceOrderWindow::getProvinceId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))//窗口没变化
                .reduce(
                        new ReduceFunction<TradeProvinceOrderWindow>() {
                            @Override
                            public TradeProvinceOrderWindow reduce(TradeProvinceOrderWindow value1,
                                                                   TradeProvinceOrderWindow value2) throws Exception {
//                                System.out.println("!!!!!!----------------");
                                value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                                value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                                value1.getUserIdSet().addAll(value2.getUserIdSet());
                                return value1;
                            }
                        },
                        new ProcessWindowFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow, String, TimeWindow>() {
                            @Override
                            public void process(String provinceId,
                                                Context context,
                                                Iterable<TradeProvinceOrderWindow> elements,
                                                Collector<TradeProvinceOrderWindow> out) throws Exception {
//                                System.out.println("??????-------------");
                                TradeProvinceOrderWindow bean = elements.iterator().next();
                                bean.setStt(AtguiguUtil.tsToDateTime(context.window().getStart()));
                                bean.setEdt(AtguiguUtil.tsToDateTime(context.window().getEnd()));

                                bean.setOrderCount((long) bean.getOrderIdSet().size());
                                bean.setOrderUserCount((long) bean.getUserIdSet().size());
                                bean.setTs(System.currentTimeMillis());
                                out.collect(bean);
                            }
                        }
                );
//        streamWithoutDim.print();

        streamWithoutDim.map(new RichMapFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow>() {
            private Connection conn;
            @Override
            public void open(Configuration parameters) throws Exception {
                conn = JdbcUtil.getPhoenixConnection();
            }

            @Override
            public TradeProvinceOrderWindow map(TradeProvinceOrderWindow bean) throws Exception {
                // 1. dim_base_province
                JSONObject skuInfo = DimUtil.readDimFromPhoenix(conn, "dim_base_province", bean.getProvinceId());
                bean.setProvinceName(skuInfo.getString("NAME"));
                 return bean;
            }
        }).addSink(FlinkSinkUtil.getClickHouseSink("dws_trade_province_order_window", TradeProvinceOrderWindow.class));
//                .print()
        ;

/*        AsyncDataStream.unorderedWait(
                        streamWithoutDim,
                        new AsyncDimFunction<TradeProvinceOrderWindow>() {
                            @Override
                            public String getTable() {
                                return "dim_base_province";
                            }

                            @Override
                            public String getId(TradeProvinceOrderWindow input) {
                                return input.getProvinceId();
                            }

                            @Override
                            public void addDim(TradeProvinceOrderWindow input, JSONObject dim) {
                                input.setProvinceName(dim.getString("NAME"));
                            }
                        },
                        120,
                        TimeUnit.SECONDS
                ).print();
              //  .addSink(FlinkSinkUtil.getClickHouseSink("dws_trade_province_order_window", TradeProvinceOrderWindow.class));*/

    }

}
